package com.hzccz.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaFileConsumer {
    private final KafkaConsumer<String, String> consumer;
    private final String topic;

    public KafkaFileConsumer(String bootstrapServers, String groupId, String topic) {
        this.topic = topic;

        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
                ConfigUtils.getProperty("consumer.auto.offset.reset", "earliest"));

        String fetchMinBytes = ConfigUtils.getProperty("consumer.fetch.min.bytes");
        if (fetchMinBytes != null) {
            properties.setProperty(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, fetchMinBytes);
        }

        String fetchMaxWaitMs = ConfigUtils.getProperty("consumer.fetch.max.wait.ms");
        if (fetchMaxWaitMs != null) {
            properties.setProperty(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, fetchMaxWaitMs);
        }

        this.consumer = new KafkaConsumer<>(properties);
        
        this.consumer.subscribe(Collections.singletonList(topic));
    }

    public void consumeAndPrint(long delayMs) throws InterruptedException {
        System.out.println("开始消费主题: " + topic);
        System.out.println("按 Ctrl+C 停止消费");

        try {
            int count = 0;
            while (true) {
                
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));

                for (ConsumerRecord<String, String> record : records) {
                    count++;
                    
                    System.out.printf("[%d] 分区: %d, 偏移量: %d, 消息: %s%n",
                            count, record.partition(), record.offset(), record.value());

                    if (delayMs > 0) {
                        Thread.sleep(delayMs);
                    }
                }

                consumer.commitAsync();
            }
        } finally {
            try {
                consumer.commitSync();
            } finally {
                consumer.close();
            }
        }
    }
}
